package top.javatool.canal.client.spring.boot.autoconfigure;

import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.*;
import top.javatool.canal.client.client.KafkaCanalClient;
import top.javatool.canal.client.factory.MapColumnModelFactory;
import top.javatool.canal.client.handler.*;
import top.javatool.canal.client.handler.impl.*;
import top.javatool.canal.client.spring.boot.properties.CanalKafkaProperties;
import top.javatool.canal.client.spring.boot.properties.CanalProperties;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;

/**
 * @author yang peng
 * @date 2019/4/1915:19
 */
@Slf4j
@Configuration
@EnableConfigurationProperties(CanalKafkaProperties.class)
@ConditionalOnBean(value = {EntryHandler.class})
@ConditionalOnProperty(value = CanalProperties.CANAL_MODE, havingValue = "kafka")
@Import(ThreadPoolAutoConfiguration.class)
public class KafkaClientAutoConfiguration {
    private CanalKafkaProperties canalKafkaProperties;

    public KafkaClientAutoConfiguration(CanalKafkaProperties canalKafkaProperties) {
        this.canalKafkaProperties = canalKafkaProperties;
    }

    @Bean
    public RowDataHandler<List<Map<String, String>>> rowDataHandler() {
        MapColumnModelFactory mapColumnModelFactory = new MapColumnModelFactory();
        MapRowDataHandlerImpl mapRowDataHandler = new MapRowDataHandlerImpl(mapColumnModelFactory);
        return mapRowDataHandler;
    }

    @Bean
    @ConditionalOnProperty(value = CanalProperties.CANAL_ASYNC, havingValue = "true", matchIfMissing = true)
    public MessageHandler messageHandler(RowDataHandler<List<Map<String, String>>> rowDataHandler, List<EntryHandler> entryHandlers, ExecutorService executorService) {
        AsyncFlatMessageHandlerImpl asyncFlatMessageHandler = new AsyncFlatMessageHandlerImpl(entryHandlers, rowDataHandler, executorService);
        return asyncFlatMessageHandler;
    }

    @Bean
    @ConditionalOnProperty(value = CanalProperties.CANAL_ASYNC, havingValue = "false")
    public MessageHandler messageHandler(RowDataHandler<List<Map<String, String>>> rowDataHandler, List<EntryHandler> entryHandlers) {
        SyncFlatMessageHandlerImpl syncFlatMessageHandler = new SyncFlatMessageHandlerImpl(entryHandlers, rowDataHandler);
        return syncFlatMessageHandler;
    }

    @Bean(initMethod = "start", destroyMethod = "stop")
    public KafkaCanalClient kafkaCanalClient(MessageHandler messageHandler) {
        log.info("初始化 KafkaCanalClient...");
        KafkaCanalClient kafkaCanalClient = KafkaCanalClient.builder()
                .servers(canalKafkaProperties.getServer())
                .groupId(canalKafkaProperties.getGroupId())
                .topic(canalKafkaProperties.getDestination())
                .messageHandler(messageHandler)
                .batchSize(canalKafkaProperties.getBatchSize())
                .filter(canalKafkaProperties.getFilter())
                .timeout(canalKafkaProperties.getTimeout())
                .unit(canalKafkaProperties.getUnit())
                .build();
        return kafkaCanalClient;
    }
}
